package com.rrd.dw.mr.orc;

import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.orc.TypeDescription;
import org.apache.orc.mapred.OrcList;
import org.apache.orc.mapred.OrcMap;
import org.apache.orc.mapred.OrcStruct;

import com.rrd.dw.utils.MD5Util;

public class ConstructUserOrcWritable {
    private TypeDescription schema = TypeDescription
        .fromString(ConstructUserMap.getSchema());
    private OrcStruct orc = (OrcStruct) OrcStruct.createValue(schema);
    private OrcList<Text> appList = (OrcList<Text>) orc.getFieldValue(54);
    private OrcMap<Text, Text> personMap = (OrcMap<Text, Text>) orc
        .getFieldValue(57);

    private OrcList<Text> userIdList = (OrcList<Text>) orc.getFieldValue(2);
    private OrcList<Text> userUniqueList = (OrcList<Text>) orc.getFieldValue(3);

    public OrcStruct getUserOrc() {
        return this.orc;
    }

    /**
     * 将业务用户输出orc
     * 
     * @param orc
     * @param mp
     */
    public OrcStruct createUserOrc(MapWritable mp) {
        String tmp = "";
        String[] arr = null;
        String[] arr2 = null;
        appList.clear();
        personMap.clear();
        String user_code = "";
        if (mp.get(new Text("mobile")) != null) {
            user_code = mp.get(new Text("mobile")).toString();
        }
        orc.setFieldValue(0,
            new Text(MD5Util.getMd5(user_code + System.currentTimeMillis())));
        orc.setFieldValue(4, (Text) mp.get(new Text("user_name")));
        orc.setFieldValue(5, (Text) mp.get(new Text("mobile")));
        orc.setFieldValue(6, (Text) mp.get(new Text("mobile_md5")));
        orc.setFieldValue(7, (Text) mp.get(new Text("mobile_off")));
        orc.setFieldValue(8, (Text) mp.get(new Text("gender")));
        orc.setFieldValue(9, mp.get(new Text("age")) == null ? null
            : new Text(etlAge(mp.get(new Text("age")).toString())));
        orc.setFieldValue(10, (Text) mp.get(new Text("edu")));
        orc.setFieldValue(11, (Text) mp.get(new Text("graduate_time")));
        orc.setFieldValue(12, (Text) mp.get(new Text("advert_job")));
        orc.setFieldValue(13, (Text) mp.get(new Text("marriage")));
        orc.setFieldValue(14, (Text) mp.get(new Text("children")));
        orc.setFieldValue(15, (Text) mp.get(new Text("industry")));
        orc.setFieldValue(16, (Text) mp.get(new Text("company_position")));
        orc.setFieldValue(17, (Text) mp.get(new Text("company_name")));
        orc.setFieldValue(18, (Text) mp.get(new Text("work_period")));
        orc.setFieldValue(19, (Text) mp.get(new Text("payday")));
        orc.setFieldValue(20, new Text(mp.get(new Text("salary")) == null ? ""
            : mp.get(new Text("salary")).toString()));
        orc.setFieldValue(21, (Text) mp.get(new Text("home_address")));
        orc.setFieldValue(22, (Text) mp.get(new Text("home_address_off")));
        orc.setFieldValue(23, (Text) mp.get(new Text("home_region")));
        orc.setFieldValue(24, (Text) mp.get(new Text("home_province")));
        orc.setFieldValue(25, (Text) mp.get(new Text("home_city")));
        orc.setFieldValue(26, (Text) mp.get(new Text("home_county")));
        orc.setFieldValue(27, (Text) mp.get(new Text("pid")));
        orc.setFieldValue(28, (Text) mp.get(new Text("pid_md5")));
        orc.setFieldValue(29, (Text) mp.get(new Text("pid_off")));
        orc.setFieldValue(30, (Text) mp.get(new Text("pid_address")));
        orc.setFieldValue(31, (Text) mp.get(new Text("pid_address_off")));
        orc.setFieldValue(32, (Text) mp.get(new Text("pid_region")));
        orc.setFieldValue(33, (Text) mp.get(new Text("pid_province")));
        orc.setFieldValue(34, (Text) mp.get(new Text("pid_city")));
        orc.setFieldValue(35, (Text) mp.get(new Text("pid_county")));
        orc.setFieldValue(36, (Text) mp.get(new Text("work_region")));
        orc.setFieldValue(37, (Text) mp.get(new Text("work_provice")));
        orc.setFieldValue(38, (Text) mp.get(new Text("work_city")));
        orc.setFieldValue(39, (Text) mp.get(new Text("work_county")));
        orc.setFieldValue(40, (Text) mp.get(new Text("ip")));
        orc.setFieldValue(41, (Text) mp.get(new Text("login_region")));
        orc.setFieldValue(42, (Text) mp.get(new Text("login_provice")));
        orc.setFieldValue(43, (Text) mp.get(new Text("login_city")));
        orc.setFieldValue(44, (Text) mp.get(new Text("login_county")));
        orc.setFieldValue(45, (Text) mp.get(new Text("regist_time")));
        orc.setFieldValue(46, (Text) mp.get(new Text("channel")));
        orc.setFieldValue(47, (Text) mp.get(new Text("source")));
        orc.setFieldValue(48, (Text) mp.get(new Text("medium")));
        orc.setFieldValue(49, (Text) mp.get(new Text("campaign")));
        orc.setFieldValue(50, (Text) mp.get(new Text("register_channel")));
        orc.setFieldValue(51, (Text) mp.get(new Text("phone_platform")));
        orc.setFieldValue(52, (Text) mp.get(new Text("device")));
        orc.setFieldValue(53, (Text) mp.get(new Text("carrier_code")));
        if (mp.get(new Text("competition_app_list")) != null) {
            tmp = mp.get(new Text("competition_app_list")).toString();
            arr = tmp.split(",");
            for (int i = 0; i < arr.length; i++) {
                appList.add(new Text(arr[i]));
            }
        }
        orc.setFieldValue(54, appList);
        orc.setFieldValue(55, (Text) mp.get(new Text("is_active")));
        orc.setFieldValue(56, (Text) mp.get(new Text("active_time")));
        if (mp.get(new Text("pre_ctact_person")) != null) {
            tmp = mp.get(new Text("pre_ctact_person")).toString();
            arr = tmp.split(",");
            for (int i = 0; i < arr.length; i++) {
                arr2 = arr[i].split("=");
                if (arr2.length == 2) {
                    personMap.put(new Text(arr2[0]), new Text(arr2[1]));
                }

            }
        }
        orc.setFieldValue(57, personMap);
        orc.setFieldValue(58, (Text) mp.get(new Text("taobao_account")));
        orc.setFieldValue(59, (Text) mp.get(new Text("email")));
        orc.setFieldValue(60, (Text) mp.get(new Text("email")));
        orc.setFieldValue(61, (Text) mp.get(new Text("email")));
        orc.setFieldValue(62, (Text) mp.get(new Text("is_del")));

//		System.out.println(orc);
        return orc;
    }

    public OrcStruct createUserMergeOrc(MapWritable[] arrMap) {
        for (int i = 1; i < arrMap.length; i++) {
            if (arrMap[i] == null) {
                continue;
            }
            changeNew(arrMap[0], arrMap[i]);
        }
        orc = createUserOrc(arrMap[0]);
        generateUserType(arrMap);
        return orc;
    }

    public void generateUserType(MapWritable[] arrMap) {
        String[] arr = new String[] { "0", "0", "0", "0", "0" };
        String[] user_id_arr = new String[] { "", "", "", "", "" };
        String[] user_unique_arr = new String[] { "", "", "", "", "" };
        String user_type = "";

        for (int i = 0; i < arrMap.length; i++) {
            if (arrMap[i] == null || arrMap[i].size() <= 0) {
                continue;
            }

            if (arrMap[i]
                .get(new Text("pdl_user_base_attribute_chain")) != null) {
                arr[4] = "1";
                user_id_arr[4] = arrMap[i].get(new Text("src_user_id"))
                    .toString();
                user_unique_arr[4] = arrMap[i].get(new Text("src_unique_id"))
                    .toString();

            } else if (arrMap[i]
                .get(new Text("rrd_user_base_attribute_chain")) != null) {
                arr[3] = "1";
                user_id_arr[3] = arrMap[i].get(new Text("src_user_id"))
                    .toString();
                user_unique_arr[3] = arrMap[i].get(new Text("src_unique_id"))
                    .toString();

            } else if (arrMap[i]
                .get(new Text("ym01_user_base_attribute_all")) != null) {
                arr[2] = "1";
                user_id_arr[2] = arrMap[i].get(new Text("src_user_id"))
                    .toString();
                user_unique_arr[2] = arrMap[i].get(new Text("src_unique_id"))
                    .toString();

            }
        }
        //生成用户ID数组
        userIdList.clear();
        userUniqueList.clear();
        for (String s : user_id_arr) {
            userIdList.add(new Text(s));
        }
        for (String s : user_unique_arr) {
            userUniqueList.add(new Text(s));
        }
        user_type = String.format("%s%s%s%s%s", arr);
        orc.setFieldValue(1, new Text(user_type));
        orc.setFieldValue(2, userIdList);
        orc.setFieldValue(3, userUniqueList);
    }

    private void changeNew(MapWritable nw, MapWritable old) {
        changeNew(nw, old, new Text("mobile"));
        changeNew(nw, old, new Text("mobile_md5"));
        changeNew(nw, old, new Text("mobile_off"));

        changeNew(nw, old, new Text("user_name"));
        changeNew(nw, old, new Text("gender"));
        changeNew(nw, old, new Text("age"));
        changeNew(nw, old, new Text("edu"));
        changeNew(nw, old, new Text("graduate_time"));
        changeNew(nw, old, new Text("advert_job"));
        changeNew(nw, old, new Text("marriage"));
        changeNew(nw, old, new Text("children"));
        changeNew(nw, old, new Text("industry"));
        changeNew(nw, old, new Text("company_position"));
        changeNew(nw, old, new Text("company_name"));
        changeNew(nw, old, new Text("payday"));
        changeNew(nw, old, new Text("salary"));
        changeNew(nw, old, new Text("home_address"));
        changeNew(nw, old, new Text("home_address_off"));
        changeNew(nw, old, new Text("home_region"));
        changeNew(nw, old, new Text("home_province"));
        changeNew(nw, old, new Text("home_city"));
        changeNew(nw, old, new Text("home_county"));
        changeNew(nw, old, new Text("pid"));
        changeNew(nw, old, new Text("pid_md5"));
        changeNew(nw, old, new Text("pid_off"));
        changeNew(nw, old, new Text("pid_address"));
        changeNew(nw, old, new Text("pid_address_off"));
        changeNew(nw, old, new Text("pid_region"));
        changeNew(nw, old, new Text("pid_province"));
        changeNew(nw, old, new Text("pid_city"));
        changeNew(nw, old, new Text("pid_county"));
        changeNew(nw, old, new Text("work_region"));
        changeNew(nw, old, new Text("work_provice"));
        changeNew(nw, old, new Text("work_city"));
        changeNew(nw, old, new Text("work_county"));
        changeNew(nw, old, new Text("ip"));
        changeNew(nw, old, new Text("login_region"));
        changeNew(nw, old, new Text("login_provice"));
        changeNew(nw, old, new Text("login_city"));
        changeNew(nw, old, new Text("login_county"));
        changeNew(nw, old, new Text("regist_time"));
        changeNew(nw, old, new Text("channel"));
        changeNew(nw, old, new Text("source"));
        changeNew(nw, old, new Text("medium"));
        changeNew(nw, old, new Text("campaign"));
        changeNew(nw, old, new Text("register_channel"));
        changeNew(nw, old, new Text("phone_platform"));
        changeNew(nw, old, new Text("device"));
        changeNew(nw, old, new Text("carrier_code"));
        changeNew(nw, old, new Text("competition_app_list"));
        changeNew(nw, old, new Text("is_active"));
        changeNew(nw, old, new Text("active_time"));
        changeNew(nw, old, new Text("pre_ctact_person"));
        changeNew(nw, old, new Text("taobao_account"));
        changeNew(nw, old, new Text("email"));
        changeNew(nw, old, new Text("email_md5"));
        changeNew(nw, old, new Text("email_off"));
        changeNew(nw, old, new Text("is_del"));
    }

    public void changeNew(MapWritable nw, MapWritable old, Text key) {
        if (nw.get(key) == null) {
            if (old.get(key) != null) {
                nw.put(key, old.get(key));
            }
        }
    }

    private String etlAge(String age) {
        if (age == null || age.equals("")) {
            return age;
        }
        int n = 0;
        try {
            n = Math.round(Float.parseFloat(age));
            if (n > 80 || n <= 0) {
                age = "";
            } else {
                age = String.valueOf(n);
            }
        } catch (Exception e) {
            age = "";
        }
        return age;
    }

}
